package com.shujia.opt

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, PreparedStatement}

object Demo2ForeachPartitions {
  def main(args: Array[String]): Unit = {
    /**
     * 将学生表的数据保存到数据库中
     */

    val conf = new SparkConf()
    //指定spark执行默认，local：本地执行
    conf.setMaster("local")
    //spark 任务名
    conf.setAppName("wc")
    //创建spark上下文对象
    val sc = new SparkContext(conf)

    //读取数据
    val studentsRDD: RDD[String] = sc.textFile("spark/data/students.csv", 10)


    /* //1、加载驱动
     Class.forName("com.mysql.jdbc.Driver")
     val start: Long = System.currentTimeMillis()
     //2、创建数据库连接
     //创建数据量连接，底层时tcp协议，需要三次握手
     //网络连接不能序列化，所以不能在网络中传输
     val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata27?useUnicode=true&characterEncoding=UTF-8&useSSL=false", "root", "123456")
     val end: Long = System.currentTimeMillis()
     println(end - start)

     /**
      * foreach: 每一条数据执行一次
      */
     studentsRDD.foreach(line => {
       //编写sql
       val stat: PreparedStatement = con.prepareStatement("insert into students values(?,?,?,?,?)")
       val stu: Array[String] = line.split(",")
       //指定字段的值
       stat.setString(1, stu(0))
       stat.setString(2, stu(1))
       stat.setInt(3, stu(2).toInt)
       stat.setString(4, stu(3))
       stat.setString(5, stu(4))
       //执行sql
       stat.execute()
       //关闭连接
       stat.close()
       con.close()
     })*/

    /**
     * foreachPartition: 一次遍历一个分区的数据
     */

    studentsRDD.foreachPartition(iter => {
      //每一个分区中创建一个连接
      //1、加载驱动
      Class.forName("com.mysql.jdbc.Driver")
      val start: Long = System.currentTimeMillis()
      //2、创建数据库连接
      //创建数据量连接，底层时tcp协议，需要三次握手
      //网络连接不能序列化，所以不能在网络中传输
      val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata27?useUnicode=true&characterEncoding=UTF-8&useSSL=false", "root", "123456")
      val end: Long = System.currentTimeMillis()
      println(end - start)
      //编写sql
      val stat: PreparedStatement = con.prepareStatement("insert into students values(?,?,?,?,?)")

      //训练一个分区中的数据
      //迭代器的foreach，不是一个算子
      iter.foreach(line => {
        val stu: Array[String] = line.split(",")
        //指定字段的值
        stat.setString(1, stu(0))
        stat.setString(2, stu(1))
        stat.setInt(3, stu(2).toInt)
        stat.setString(4, stu(3))
        stat.setString(5, stu(4))
        //执行sql
        stat.execute()
      })
      //关闭连接
      stat.close()
      con.close()
    })
  }
}
